/*
 * yNetServer.cpp
 *
 *  Created on: Sep 13, 2016
 *      Author: root
 */
#include "yNetServer.h"
#include "yByteTools.h"
NetServer * NetServer::instance = NULL;
NetServer::NetServer(int p) {
	port = p;
	shut_server = 0;
	lisSock = 0;
	dispatcher = NULL;
	NetServer::instance = this;
}

NetServer::~NetServer() {

}

int NetServer::start() {
	int c;
	for (c = 0; c < CONN_MAXFD; ++c) {
		g_conn_table[c].sock = c;
		g_conn_table[c].data = NULL;
	}
	struct sigaction act;
	memset(&act, 0, sizeof(act));

	act.sa_handler = &NetServer::shutServerHandler;
	sigaction(SIGINT, &act, NULL);
	sigaction(SIGTERM, &act, NULL);

	/* create 2 different epoll fd */
	int epi;
	for (epi = 0; epi < EPOLL_NUM; ++epi) {
		epfd[epi] = epoll_create(20);
	}

	lisSock = socket(AF_INET, SOCK_STREAM, 0);

	int reuse = 1;
	setsockopt(lisSock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));

	int flag;
	flag = fcntl(lisSock, F_GETFL);
	fcntl(lisSock, F_SETFL, flag | O_NONBLOCK);

	struct sockaddr_in lisAddr;
	lisAddr.sin_family = AF_INET;
	lisAddr.sin_port = htons(port);
	lisAddr.sin_addr.s_addr = htonl(INADDR_ANY);

	if (bind(lisSock, (struct sockaddr *) &lisAddr, sizeof(lisAddr)) == -1) {
		perror("bind");
		return -1;
	}

	listen(lisSock, 4096);

	pthread_t lisTid;
	pthread_create(&lisTid, NULL, &NetServer::createListenThread, this);

	int i;

	for (i = 0; i < EPOLL_NUM; ++i) {
		int j;
		for (j = 0; j < WORKER_PER_GROUP; ++j) {
			pthread_create(worker + (i * WORKER_PER_GROUP + j), NULL, &NetServer::createWorkerThread, epfd + i);
		}
	}

	for (i = 0; i < NUM_WORKER; ++i) {
		pthread_join(worker[i], NULL);
	}

	pthread_join(lisTid, NULL);

	struct epoll_event evReg;

	for (c = 0; c < CONN_MAXFD; ++c) {
		connection_t conn = g_conn_table + c;
		if (conn->inuse) {
			epoll_ctl(epfd[conn->index], EPOLL_CTL_DEL, conn->sock, &evReg);
			close(conn->sock);
		}
	}

	for (epi = 0; epi < EPOLL_NUM; ++epi) {
		close(epfd[epi]);
	}
	close(lisSock);

	return 0;
}

void NetServer::shutServerHandler(int signo) {
	NetServer::getInstance()->shut_server = 1;
}

int NetServer::sendData(connection_t conn, char* data, int len) {
	if (conn->woff) {
		if (conn->woff + len > BUF_SIZE) {
			return -1;
		}
		memcpy(conn->wbuf + conn->woff, data, len);
		conn->woff += len;
		return 0;
	} else {
		int ret = write(conn->sock, data, len);
		if (ret > 0) {
			if (ret == len) {
				return 0;
			}
			int left = len - ret;
			if (left > BUF_SIZE)
				return -1;

			memcpy(conn->wbuf, data + ret, left);
			conn->woff = left;
		} else {
			if (errno != EINTR && errno != EAGAIN) {
				return -1;
			}
			if (len > BUF_SIZE) {
				return -1;
			}
			memcpy(conn->wbuf, data, len);
			conn->woff = len;
		}
	}

	return 0;
}

int NetServer::handleReadEvent(connection_t conn) {

	int ret = read(conn->sock, conn->rbuf + conn->roff, BUF_SIZE - conn->roff);

	if (ret > 0) {
		conn->roff += ret;
		if (conn->roff < 4)
			return 0;
		int len = read_int(conn->rbuf);
		if (conn->roff < len)
			return 0;
		char * buf = (char*) malloc(len - 4);
		memcpy(buf, conn->rbuf + 4, len - 4);
		memmove(conn->rbuf, conn->rbuf + len, conn->roff - len);
		conn->roff -= len;
		dispatcher->handle(conn, buf, len - 4);
	} else if (ret == 0) {
		return -1;
	} else {
		if (errno != EINTR && errno != EAGAIN) {
			return -1;
		}
	}

	return 0;
}

int NetServer::handleWriteEvent(connection_t conn) {
	if (conn->woff == 0)
		return 0;

	int ret = write(conn->sock, conn->wbuf, conn->woff);

	if (ret == -1) {
		if (errno != EINTR && errno != EAGAIN) {
			return -1;
		}
	} else {
		int left = conn->woff - ret;

		if (left > 0) {
			memmove(conn->wbuf, conn->wbuf + ret, left);
		}

		conn->woff = left;
	}

	return 0;
}

void NetServer::closeConnection(connection_t conn) {
	struct epoll_event evReg;

	conn->inuse = 0;
	conn->woff = conn->roff = 0;
	epoll_ctl(epfd[conn->index], EPOLL_CTL_DEL, conn->sock, &evReg);
	close(conn->sock);
}

void* NetServer::createWorkerThread(void* arg) {
	int epfd = *(int *) arg;

	struct epoll_event event;
	struct epoll_event evReg;
	NetServer* server = NetServer::getInstance();
	/* only handle connected socket */
	while (!server->shut_server) {
		int numEvents = epoll_wait(epfd, &event, 1, 1000);

		if (numEvents > 0) {
			int sock = event.data.fd;
			connection_t conn = &server->g_conn_table[sock];

			if (event.events & EPOLLOUT) {
				if (server->handleWriteEvent(conn) == -1) {
					server->closeConnection(conn);
					continue;
				}
			}

			if (event.events & EPOLLIN) {
				if (server->handleReadEvent(conn) == -1) {
					server->closeConnection(conn);
					continue;
				}
			}

			evReg.events = EPOLLIN | EPOLLONESHOT;
			if (conn->woff > 0)
				evReg.events |= EPOLLOUT;
			evReg.data.fd = sock;
			epoll_ctl(epfd, EPOLL_CTL_MOD, conn->sock, &evReg);
		}
	}
	return NULL;
}

void* NetServer::createListenThread(void* arg) {
	int lisEpfd = epoll_create(5);
	NetServer* server = NetServer::getInstance();
	struct epoll_event evReg;
	evReg.events = EPOLLIN;
	evReg.data.fd = server->lisSock;

	epoll_ctl(lisEpfd, EPOLL_CTL_ADD, server->lisSock, &evReg);

	struct epoll_event event;

	int rrIndex = 0; /* round robin index */

	/* only handle listen socekt */
	while (!server->shut_server) {
		int numEvent = epoll_wait(lisEpfd, &event, 1, 1000);
		if (numEvent > 0) {
			int sock = accept(server->lisSock, NULL, NULL);
			if (sock > 0) {
				server->g_conn_table[sock].inuse = 1;

				int flag;
				flag = fcntl(sock, F_GETFL);
				fcntl(sock, F_SETFL, flag | O_NONBLOCK);

				evReg.data.fd = sock;
				evReg.events = EPOLLIN | EPOLLONESHOT;

				/* register to worker-pool's epoll,
				 * not the listen epoll */
				server->g_conn_table[sock].index = rrIndex;
				server->g_conn_table[sock].data = NULL;
				epoll_ctl(server->epfd[rrIndex], EPOLL_CTL_ADD, sock, &evReg);
				rrIndex = (rrIndex + 1) % EPOLL_NUM;
			}
		}
	}

	close(lisEpfd);
	return NULL;
}

NetServer* NetServer::getInstance() {
	return instance;
}

void NetServer::handleMsg(context ctx, char* buf, int len) {
}
